<?php
/**
 * 事务处理
 * User: dxk <1024114660@qq.com>
 * Date: 2021-03-04
 */
namespace App\Library\Service\DB;

use App\Library\ThirdParty\DingTalk\DingTalkSendMsg;
use Illuminate\Database\Events\TransactionBeginning;
use Illuminate\Database\Events\TransactionCommitted;
use Illuminate\Database\Events\TransactionRolledBack;

class TransactionMonitorService
{

    /**
     * 请求结束，检查事务是否闭合的监控，检查事务期间的缓存
     * @param array $request
     * @throws \Exception
     */
    public static function requestEnd(array $request)
    {
        self::execTransFunc(true);
        self::execTransCache(true);
        global $IS_WEB_ERROR;
        global $SQL_LOG;
        global $SQL_TRANSACTION;
        //----事务未闭合判断
        if (!empty($SQL_TRANSACTION) && !$IS_WEB_ERROR) {
            foreach ($SQL_TRANSACTION as $item) {
                if ($item != 0) {
                    $msg = $request;
                    $msg['action'] = '事务未闭合,请及时处理';
                    DingTalkSendMsg::sendText($msg);
                    $msg['sql_transaction'] = $SQL_TRANSACTION;
                    $msg['sql_log'] = $SQL_LOG;
                    easy_log('trans_error', $msg);
                    break;
                }
            }
        }
        //-----跨库事务检测
        global $SQL_TRANSACTION_TRACE;
        if (!empty($SQL_TRANSACTION_TRACE) && is_array($SQL_TRANSACTION_TRACE) && !$IS_WEB_ERROR) {
            $trans = [];
            $trans_trace = [];
            foreach ($SQL_TRANSACTION_TRACE as $item) {
                if ($item['action'] == 'start') {
                    $trans[$item['conn']]++;
                } else {
                    $trans[$item['conn']] -= 1;
                    $trans_trace[$item['conn']] = $item['action'];
                    if ($trans[$item['conn']] == 0) {
                        unset($trans[$item['conn']]);
                        if (empty($trans)) {
                            if (count($trans_trace) > 1) {
                                $trans_trace = array_unique($trans_trace);
                                if (count($trans_trace) > 1) {
                                    $msg = $request;
                                    $msg['action'] = '严重的错误，跨库事务异常，请及时处理';
                                    DingTalkSendMsg::sendText($msg);
                                    $msg['sql_transaction_trace'] = $SQL_TRANSACTION_TRACE;
                                    $msg['sql_transaction'] = $SQL_TRANSACTION;
                                    $msg['sql_log'] = $SQL_LOG;
                                    easy_log('trans_error', $msg);
                                    break;
                                }
                            }
                            $trans_trace = [];
                        }
                    }
                }
            }
        }
        //关闭事务,防止队列中事务没关闭造成异常
        if (!empty($SQL_TRANSACTION) && is_array($SQL_TRANSACTION)) {
            $not_close = false;
            foreach ($SQL_TRANSACTION as $connectionName => $item) {
                if ($item != 0) {
                    \DB::disconnect($connectionName);
                    $not_close = true;
                }
            }
            if ($not_close) {
                easy_log('close', [$request, $SQL_TRANSACTION, $SQL_LOG, 'is_job' => defined('IS_JOB') ? IS_JOB : 0]);
            }
        }
        $IS_WEB_ERROR = false;
        $SQL_LOG = [];
        $SQL_TRANSACTION = [];
        $SQL_TRANSACTION_TRACE = [];
    }

    /**
     * 检测是否在执行事务
     * @return bool
     */
    public static function isTransaction(): bool
    {
        global $SQL_TRANSACTION;
        if (empty($SQL_TRANSACTION)) {
            return false;
        }
        foreach ($SQL_TRANSACTION as $item) {
            if ($item != 0) {
                return true;
            }
        }
        return false;
    }


    /**
     * 主要用于日志写入，防止事务回滚对数据插入造成影响
     * @param \Closure $obj
     * @return array|mixed
     */
    public static function transFunc(\Closure $obj): mixed
    {
        if (self::isTransaction()) {
            global $TRANSACTION_FUNC;
            $TRANSACTION_FUNC[] = $obj;
            return alert_info(1, '等待事务结束后执行');
        } else {
            return $obj();
        }
    }

    /**
     * 异步方法处理
     * @return void
     */
    public static function execTransFunc($is_force = false)
    {
        if ($is_force || !self::isTransaction()) {
            global $TRANSACTION_FUNC;
            if (is_array($TRANSACTION_FUNC)) {
                foreach ($TRANSACTION_FUNC as $k => $item) {
                    $item();
                    unset($TRANSACTION_FUNC[$k]);
                }
            }
        }
    }

    /**
     * 事务缓存处理
     * @return void
     */
    public static function execTransCache($is_force = false)
    {
        if ($is_force || !self::isTransaction()) {
            global $TRANSACTION_CACHE;
            if (is_array($TRANSACTION_CACHE)) {
                foreach ($TRANSACTION_CACHE as $k => $item) {
                    $cache_instance = $item['cache_instance'];
                    foreach ($item['cache'] as $cache_k) {
                        $cache_instance->forget($cache_k);
                    }
                    unset($TRANSACTION_CACHE[$k]);
                }
            }
        }
    }

    /**
     * 事务缓存
     * @param $instance
     * @param $cache_k
     */
    public static function transCache($instance, $cache_k)
    {
        if (self::isTransaction()) {
            global $TRANSACTION_CACHE;
            $TRANSACTION_CACHE[md5(json_encode($instance))]['cache_instance'] = $instance;
            $TRANSACTION_CACHE[md5(json_encode($instance))]['cache'][$cache_k] = $cache_k;
        }
    }

    /**
     * 开启事务
     * @param TransactionBeginning $event
     */
    public static function begin(TransactionBeginning $event)
    {
        global $SQL_TRANSACTION;
        $SQL_TRANSACTION[$event->connectionName]++;
        if (defined('IS_WEB') || defined('IS_JOB')) {
            global $SQL_TRANSACTION_TRACE;
            $SQL_TRANSACTION_TRACE[] = ['conn' => $event->connectionName, 'action' => 'start'];
            if (env('SQL_LOG') !== 0) {
                global $SQL_LOG;
                $SQL_LOG[] = ['sql' => '开启事务', 'connectionName' => $event->connectionName];
            }
        }
    }

    /**
     * 回滚事务
     * @param TransactionRolledBack $event
     */
    public static function rollback(TransactionRolledBack $event)
    {
        global $SQL_TRANSACTION;
        $SQL_TRANSACTION[$event->connectionName] -= 1;
        if (defined('IS_WEB') || defined('IS_JOB')) {
            global $SQL_TRANSACTION_TRACE;
            $SQL_TRANSACTION_TRACE[] = ['conn' => $event->connectionName, 'action' => 'rollback'];
            if (env('SQL_LOG') !== 0) {
                global $SQL_LOG;
                $SQL_LOG[] = ['sql' => '回滚事务', 'connectionName' => $event->connectionName];
            }
        }
        self::execTransFunc();
        self::execTransCache();
    }

    /**
     * 提交事务
     * @param TransactionCommitted $event
     */
    public static function commit(TransactionCommitted $event)
    {
        global $SQL_TRANSACTION;
        $SQL_TRANSACTION[$event->connectionName] -= 1;
        if (defined('IS_WEB') || defined('IS_JOB')) {
            global $SQL_TRANSACTION_TRACE;
            $SQL_TRANSACTION_TRACE[] = ['conn' => $event->connectionName, 'action' => 'commit'];
            if (env('SQL_LOG') !== 0) {
                global $SQL_LOG;
                $SQL_LOG[] = ['sql' => '提交事务', 'connectionName' => $event->connectionName];
            }
        }
        self::execTransFunc();
        self::execTransCache();
    }
}
